/*
 * Copyright (c) 2006-2021, RT-Thread Development Team
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * Change Logs:
 * Date           Author       Notes
 * 2022-05-29     vandoul       the first version
 */

#include <rtthread.h>

#include "paho_mqtt.h"

#include "srv_mqtt.h"
#include "user_config.h"

#define LOG_LVL         LOG_LVL_DBG
#define LOG_TAG         "SRV_MQTT"
#include "ulog.h"

#define SRV_MQTT_EVENT_REQ_READY            1
#define SRV_MQTT_EVENT_RSP_READY            2

struct srv_mqtt_object {
    int id;
    char name[16];
    rt_bool_t is_init;
    rt_bool_t is_connected;
    MQTTClient client;
    rt_thread_t tid;
    rt_event_t srv_evt;
    rt_mutex_t srv_lock;
    struct srv_mqtt_data *data;
    char url[64];
    char prefix[64];
};

static struct srv_mqtt_object srv_mqtt_obj[SRV_MQTT_MAX_NUM];

static void mqtt_sub_callback(MQTTClient *c, MessageData *msg_data)
{
    *((char *)msg_data->message->payload + msg_data->message->payloadlen) = '\0';
    LOG_D("mqtt sub callback: %.*s %.*s",
               msg_data->topicName->lenstring.len,
               msg_data->topicName->lenstring.data,
               msg_data->message->payloadlen,
               (char *)msg_data->message->payload);
}
static void mqtt_sub_default_callback(MQTTClient *c, MessageData *msg_data)
{
    *((char *)msg_data->message->payload + msg_data->message->payloadlen) = '\0';
    LOG_D("mqtt sub default callback: %.*s %.*s",
               msg_data->topicName->lenstring.len,
               msg_data->topicName->lenstring.data,
               msg_data->message->payloadlen,
               (char *)msg_data->message->payload);
}
static void mqtt_connect_callback(MQTTClient *c)
{
    LOG_D("inter mqtt_connect_callback!");
}
static void mqtt_online_callback(MQTTClient *c)
{
    struct srv_mqtt_object *mqtt_obj = (struct srv_mqtt_object *)c->user_data;
    LOG_D("inter mqtt_online_callback!");
    mqtt_obj->is_connected = RT_TRUE;
}
static void mqtt_offline_callback(MQTTClient *c)
{
    struct srv_mqtt_object *mqtt_obj = (struct srv_mqtt_object *)c->user_data;
    mqtt_obj->is_connected = RT_FALSE;
    LOG_D("inter mqtt_offline_callback!");
}

static void srv_mqtt_task(void *args)
{
    char topic[128];
    struct srv_mqtt_data *data;
    struct srv_mqtt_object *mqtt_obj = (struct srv_mqtt_object *)args;
    MQTTClient *client = &mqtt_obj->client;
    paho_mqtt_start(client);
    LOG_D("%s mqtt task run.", mqtt_obj->name);
    while(1) {
        if(RT_EOK == rt_event_recv(mqtt_obj->srv_evt, SRV_MQTT_EVENT_REQ_READY, RT_EVENT_FLAG_OR|RT_EVENT_FLAG_CLEAR, RT_WAITING_FOREVER, RT_NULL)) {
            data = mqtt_obj->data;
            if(data) {
                MQTTMessage message;
                if(mqtt_obj->is_connected) {
                    message.qos = data->qos;
                    message.retained = 0;
                    message.payload = data->buff;
                    message.payloadlen = data->size;
                    rt_sprintf(topic, "%s%s/data", mqtt_obj->prefix, data->topic);
                    LOG_D("%s topic:%s, payload(%d):%s", mqtt_obj->name, topic, data->size, data->buff);
                    if(0 != MQTTPublish(client, topic, &message)) {
                        data->rsp_num = -1;
                    } else {
                        data->rsp_num = 0;
                    }
//                    if(0 != paho_mqtt_publish(client, data->qos, data->topic, data->buff)) {
//                        data->rsp_num = -1;
//                    } else {
//                        data->rsp_num = 0;
//                    }
                } else {
                    data->rsp_num = -1;
                }
            } else {
                LOG_E("%s data == RT_NULL", mqtt_obj->name);
            }
            rt_event_send(mqtt_obj->srv_evt, SRV_MQTT_EVENT_RSP_READY);
        } else {
            LOG_E("%s wait event failed", mqtt_obj->name);
        }
    }
}
int srv_mqtt_create(struct srv_mqtt_config *cfg)
{
    const static MQTTPacket_connectData condata = MQTTPacket_connectData_initializer;
    struct srv_mqtt_object *mqtt_obj = RT_NULL;
    int index;
    for(index = 0; index < SRV_MQTT_MAX_NUM; index++) {
        mqtt_obj = &srv_mqtt_obj[index];
        if(mqtt_obj->is_init) {
            continue;
        }
        mqtt_obj->is_init = RT_TRUE;
        mqtt_obj->id = index;
        mqtt_obj->is_connected = RT_FALSE;
        break;
    }
    if(index < SRV_MQTT_MAX_NUM) {
        MQTTClient *client = &mqtt_obj->client;
        rt_err_t ret = RT_EOK;
        char name[16];
        int len = rt_strlen(cfg->url);
//        LOG_D("id:%d,name:%s,url:%s,prefix:%s", index, cfg->name, cfg->url, cfg->prefix);
        rt_memcpy(mqtt_obj->url, cfg->url, len);
        mqtt_obj->url[len] = '\0';
        len = rt_strlen(cfg->prefix);
        rt_memcpy(mqtt_obj->prefix, cfg->prefix, len);
        mqtt_obj->prefix[len] = '\0';
        len = rt_strlen(cfg->name);
        rt_memcpy(mqtt_obj->name, cfg->name, len);
        mqtt_obj->name[len] = '\0';
        do {
            rt_sprintf(name, "mqevt%d", mqtt_obj->id);
            mqtt_obj->srv_evt = rt_event_create(name, RT_IPC_FLAG_FIFO);
            if(mqtt_obj->srv_evt == RT_NULL) {
                LOG_E("create event failed");
                break;
            }
            rt_sprintf(name, "mqlock%d", mqtt_obj->id);
            mqtt_obj->srv_lock = rt_mutex_create(name, RT_IPC_FLAG_FIFO);
            if(mqtt_obj->srv_lock == RT_NULL) {
                LOG_E("create lock failed");
                break;
            }

            client->isconnected = 0;
            client->uri = mqtt_obj->url;

            /* config connect param */
            rt_memcpy(&client->condata, &condata, sizeof(condata));
            client->condata.clientID.cstring = mqtt_obj->name;
            client->condata.keepAliveInterval = 30;
            client->condata.cleansession = 1;
//            client->condata.username.cstring = RT_NULL;
//            client->condata.password.cstring = RT_NULL;

            /* config MQTT will param. */
//            client->condata.willFlag = 1;
//            client->condata.will.qos = 1;
//            client->condata.will.retained = 0;
//            client->condata.will.topicName.cstring = MQTT_PUBTOPIC;
//            client->condata.will.message.cstring = MQTT_WILLMSG;

            /* malloc buffer. */
            client->buf_size = client->readbuf_size = 1024;
            client->buf = rt_calloc(1, client->buf_size);
            client->readbuf = rt_calloc(1, client->readbuf_size);
            if (!(client->buf && client->readbuf))
            {
                LOG_E("no memory for MQTT client buffer!");
                break;
            }

            /* set event callback function */
            client->connect_callback = mqtt_connect_callback;
            client->online_callback = mqtt_online_callback;
            client->offline_callback = mqtt_offline_callback;

            /* set subscribe table and event callback */
//            client->messageHandlers[0].topicFilter = rt_strdup(MQTT_SUBTOPIC);
//            client->messageHandlers[0].callback = mqtt_sub_callback;
//            client->messageHandlers[0].qos = QOS1;

            /* set default subscribe event callback */
            client->defaultMessageHandler = mqtt_sub_default_callback;

            client->user_data = mqtt_obj;

            LOG_D("id:%d,name:%s,url:%s,prefix:%s", mqtt_obj->id, mqtt_obj->name, mqtt_obj->url, mqtt_obj->prefix);

            mqtt_obj->tid = rt_thread_create(mqtt_obj->name, srv_mqtt_task, mqtt_obj, 2048, 6, 10);
            if(mqtt_obj->tid == RT_NULL) {
                LOG_E("create task failed");
                break;
            }
            return RT_EOK;
        } while(0);
        if(mqtt_obj->tid) {
            rt_thread_delete(mqtt_obj->tid);
            mqtt_obj->tid = RT_NULL;
        }
        if(mqtt_obj->srv_evt) {
            rt_event_delete(mqtt_obj->srv_evt);
            mqtt_obj->srv_evt = RT_NULL;
        }
        if(mqtt_obj->srv_lock) {
            rt_mutex_delete(mqtt_obj->srv_lock);
            mqtt_obj->srv_lock = RT_NULL;
        }
        if(mqtt_obj->is_init) {
            mqtt_obj->is_init = RT_FALSE;
        }
    }
    return -RT_ERROR;
}
int srv_mqtt_start(int srv_num)
{
    if((srv_num >= 0) && (srv_num < SRV_MQTT_MAX_NUM)) {
        struct srv_mqtt_object *mqtt_obj = &srv_mqtt_obj[srv_num];
        if(mqtt_obj->tid) {
            if(rt_thread_startup(mqtt_obj->tid) ==RT_EOK) {
                return 0;
            }
        }
    }
    return -1;
}
int srv_mqtt_trans(int srv_num, struct srv_mqtt_data *data)
{
    if(data == RT_NULL) {
        return -1;
    }
    if((srv_num >= 0) && (srv_num < SRV_MQTT_MAX_NUM)) {
        struct srv_mqtt_object *mqtt_obj = &srv_mqtt_obj[srv_num];
        rt_mutex_take(mqtt_obj->srv_lock, RT_WAITING_FOREVER);
        mqtt_obj->data = data;
        rt_event_send(mqtt_obj->srv_evt, SRV_MQTT_EVENT_REQ_READY);
        rt_event_recv(mqtt_obj->srv_evt, SRV_MQTT_EVENT_RSP_READY, RT_EVENT_FLAG_OR|RT_EVENT_FLAG_CLEAR, RT_WAITING_FOREVER, RT_NULL);
        rt_mutex_release(mqtt_obj->srv_lock);
        return 0;
    }
    return -1;
}
int srv_mqtt_get_srv_num(char *name)
{
    if(name == RT_NULL) {
        return -1;
    }
    for(int i=0; i<SRV_MQTT_MAX_NUM; i++) {
        if(rt_strcmp(name, srv_mqtt_obj[i].name) == 0) {
            return i;
        }
    }
    return -1;
}
/******************************************************/
#include <finsh.h>
static uint8_t srv_mqtt_test_buff[512];
static void srv_mqtt_test_task(void *args)
{
    rt_tick_t last_tick;
    struct srv_mqtt_data data;
    LOG_D("mqtt test run.");
    srv_mqtt_start(0);
    data.qos = 1;
    data.topic = "/device/nuc980/data";
    data.buff = srv_mqtt_test_buff;
    while(1) {
        last_tick = rt_tick_get();
        data.size = rt_sprintf(srv_mqtt_test_buff, "{\"cur_tick\":%d}", rt_tick_get());
        if(0 == srv_mqtt_trans(0, &data)) {
            LOG_D("rsp_num:%d", data.rsp_num);
        } else {
            LOG_E("srv mqtt trans failed!");
        }
        rt_thread_delay_until(&last_tick, rt_tick_from_millisecond(2000));
    }
}
int srv_mqtt_test(int argc, char *argv[])
{
    struct srv_mqtt_config cfg = {
        /* .name = */{"mqtt_test"},
        /* .url  = */"tcp://223.95.85.133:1883"
    };
    srv_mqtt_create(&cfg);
    rt_thread_t tid = rt_thread_create("mqtt_test", srv_mqtt_test_task, RT_NULL, 2048, 6, 10);
    if(tid) {
        rt_thread_startup(tid);
    }
    return 0;
}
MSH_CMD_EXPORT(srv_mqtt_test, srv mqtt test.);
